package cn.daomeng.firebird.mq.rabbit;

import java.util.HashSet;
import java.util.Set;
import java.util.UUID;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;

/**
 * rabbitService消息模块
 * @author qsl
 * @date 2019.05.13
 */
public class SimpleRabbitService implements RabbitMQService {

    private static final Logger logger = LoggerFactory.getLogger(SimpleRabbitService.class);

    protected static final long DEFAUL_SLEEP_MILLIS = 30;

    private ConnectionFactory rabbitMQConnectionFactory;

    private RabbitTemplate rabbitTemplate;

    private RabbitAdmin rabbitAdmin;

    private Set<String> declaredQueues = new HashSet<String>();

    private Set<String> declaredExchangeAndQueues = new HashSet<String>();

    private MessageConverter messageConverter;

    public SimpleRabbitService() {
    }

    public SimpleRabbitService(ConnectionFactory cf, RabbitTemplate rt, RabbitAdmin admin) {
        this(cf, rt, admin, null);
    }

    public SimpleRabbitService(ConnectionFactory cf, RabbitTemplate rt, RabbitAdmin admin, MessageConverter mc) {
        this.rabbitMQConnectionFactory = cf;
        this.rabbitTemplate = rt;
        this.rabbitAdmin = admin;
        this.messageConverter = mc;
        if (this.messageConverter != null) {
            rabbitTemplate.setMessageConverter(messageConverter);
        }
    }

    /**
     * 信息发送(exchangeType:direct)
     *
     * @param queueName queue名称
     * @param data      发送数据对象
     */
    @Override
    public void send(String queueName, Object data) {
        send(queueName, queueName, ExchangeType.DIRECT, data, new String[]{queueName});
    }

    /**
     * 信息发送
     *
     * @param target 消息类型对象，详见{@RabbitMQMessageTarget}
     * @param data                  数据对象
     */
    @Override
    public void send(RabbitMQMessageTarget target, Object data) {
        send(target.getExchangeName(), target.getRoutingKey(), target.getExchangeType(), data, target.getQueueNames());
    }

    protected void send(String exchangeName, String routingKey, ExchangeType exchangeType, Object data,
                        String... queueNames) {
        if (StringUtils.isEmpty(exchangeName)) {
            throw new IllegalArgumentException("exchange or routingKey must not be null");
        }
        // 创建queue，exchange，并创建绑定关系
        this.declareExchangeAndQueue(exchangeName, exchangeType, routingKey, queueNames);
        try {

            Message message = null;

            MessageProperties messageProperties = new MessageProperties();
            messageProperties.setMessageId(UUID.randomUUID().toString());
            if(messageConverter == null){
                SimpleMessageConverter defaultConverter = new SimpleMessageConverter();
                message = defaultConverter.toMessage(data, messageProperties);
            }else{
                message = messageConverter.toMessage(data, messageProperties);
            }

            rabbitTemplate.convertAndSend(exchangeName, routingKey, message);

        } catch (AmqpException e) {
            logger.error("RabbitMQ send exception" + e.getMessage(), e);
            throw e;
        }
    }

    private void declareExchangeAndQueue(String exchangeName, ExchangeType ExchangeType, String routingKey,
                                         String... queueNames) {
        if (queueNames != null && queueNames.length > 0) {
            for (String queueName : queueNames) {
                if (!declaredExchangeAndQueues.contains(exchangeName + "|" + queueName)) {
                    Queue queue = new Queue(queueName);
                    queue.setAdminsThatShouldDeclare(rabbitAdmin);
                    rabbitAdmin.declareQueue(queue);


                    String deadQueueName = queueName+"_DEAD";
                    Queue deadQueue = new Queue(deadQueueName);
                    deadQueue.setAdminsThatShouldDeclare(rabbitAdmin);
                    rabbitAdmin.declareQueue(deadQueue);

                    switch (ExchangeType) {
                        case TOPIC:
                            TopicExchange topicExchange = new TopicExchange(exchangeName);
                            rabbitAdmin.declareExchange(topicExchange);
                            rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(topicExchange).with(routingKey));
                            break;
                        case DIRECT:
                            DirectExchange directExchange = new DirectExchange(exchangeName);
                            rabbitAdmin.declareExchange(directExchange);
                            rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(directExchange).with(routingKey));

                            DirectExchange deadDirectExchange = new DirectExchange(exchangeName);
                            rabbitAdmin.declareExchange(deadDirectExchange);
                            rabbitAdmin.declareBinding(BindingBuilder.bind(deadQueue).to(deadDirectExchange).with(deadQueueName));

                            break;
                        case FANOUT:
                            FanoutExchange fanoutExchange = new FanoutExchange(exchangeName);
                            rabbitAdmin.declareExchange(fanoutExchange);
                            rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(fanoutExchange));
                            break;
                        default:
                            FanoutExchange exchange = new FanoutExchange(exchangeName);
                            rabbitAdmin.declareExchange(exchange);
                            rabbitAdmin.declareBinding(BindingBuilder.bind(queue).to(exchange));
                            break;
                    }
                    declaredExchangeAndQueues.add(exchangeName + "|" + queueName);
                }
            }
        }
    }

    protected enum ExchangeType {
        TOPIC, DIRECT, FANOUT
    }


    /**
     * 注册listener
     * @param listener
     * @param concurrentConsumers 监听者的个数
     */
    @Override
    public void listen(final RabbitMessageListener listener, Integer concurrentConsumers) {
        String targetQueue = listener.getTargetQueueName();
        this.ensureQueueDeclared(targetQueue);


        ChannelAwareMessageListener messageHandler = new ChannelAwareMessageListener(){
            @Override
            public void onMessage(Message message, Channel channel) throws Exception{
                String consumerQueue = message.getMessageProperties().getConsumerQueue();
                Integer count = (Integer) message.getMessageProperties().getHeaders().get("count");

                Object object = null;
                if(messageConverter == null){
                    SimpleMessageConverter defaultConverter = new SimpleMessageConverter();
                    object = defaultConverter.fromMessage(message);
                }else{
                    object = messageConverter.fromMessage(message);
                }
                // 如果是手动ack
                if(listener.getAcknowledgeMode() == RabbitMessageListener.ACK_MODE_MANUAL){
                // 如果是无需ack
                    listener.handleMessage(object, message , channel);
                }else if (listener.getAcknowledgeMode() == RabbitMessageListener.ACK_MODE_NONE){
                    listener.handleMessage(object);
                }else{
                    // 如果是自动ack
                    if(count == null){
                        try {
                            listener.handleMessage(object);
                        } catch (Exception e) {
                            //原来的消息手动ACK，然后发送信的消息，消息体一样，增加header头count = 1
                            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                            message.getMessageProperties().setHeader("count",1);
                            rabbitTemplate.convertAndSend(consumerQueue, consumerQueue, message);
                            logger.error("MQ listener handle method exception " + e.getMessage(), e);
                        }
                    }else if(count > 0 && count <= 5){
                        try {
                            listener.handleMessage(object);
                        } catch (Exception e) {
                            //原来的消息手动ACK，然后发送信的消息，消息体一样，增加header头count = 1
                            channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                            message.getMessageProperties().setHeader("count",count + 1);
                            rabbitTemplate.convertAndSend(consumerQueue, consumerQueue, message);
                            logger.error("MQ listener handle method exception " + e.getMessage(), e);
                        }

                    }else if(count > 5){
                        //加入到死信队列中
                        String deadQueue = consumerQueue + "_DEAD";
                        rabbitTemplate.convertAndSend(deadQueue, deadQueue, message);
                        //确认消息已经消费，然后需要查询嗯时候，从私信队列中消费
                        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
                        System.out.println("消息已经加入到私信队列-" +deadQueue + "-" + JSON.toJSONString(object));
                    }
                }
            }
        };
        // 注册监听接口
        MessageListenerAdapter adapter = new MessageListenerAdapter(messageHandler);

        if (messageConverter != null) {
            adapter.setMessageConverter(messageConverter);
        }

        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(rabbitMQConnectionFactory);
        container.setMessageListener(adapter);
        container.setQueueNames(listener.getTargetQueueName());
        if (concurrentConsumers != null && concurrentConsumers.intValue() > 1) {
            container.setConcurrentConsumers(concurrentConsumers);
        }
        switch (listener.getAcknowledgeMode()) {
            case RabbitMessageListener.ACK_MODE_AUTO: {
                container.setAcknowledgeMode(AcknowledgeMode.AUTO);
                break;
            }
            case RabbitMessageListener.ACK_MODE_MANUAL: {
                container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
                break;
            }
            case RabbitMessageListener.ACK_MODE_NONE: {
                container.setAcknowledgeMode(AcknowledgeMode.NONE);
                break;
            }
            default: {
                container.setAcknowledgeMode(AcknowledgeMode.AUTO);
                break;
            }
        }

        container.start();
    }

    /**
     * 声明exchange与queue
     *
     * @param queueName
     */
    private void ensureQueueDeclared(String queueName) {
        if (!declaredQueues.contains(queueName)) {
            Queue queue = new Queue(queueName);
            queue.setAdminsThatShouldDeclare(rabbitAdmin);
            rabbitAdmin.declareQueue(queue);

            declaredQueues.add(queueName);
        }
    }
}
